

package SSF.OS;

import SSF.Util.Streams.*;
import com.renesys.raceway.DML.*;
import java.util.*;

/** Class facilitating the use of the package SSF.Util.Streams.
 *  From any protocol or protocol-related code, access the "probe"
 *  protocol and call <code>getRecorder</code> to get a handle on an
 *  implementation of <code>StreamInterface</code> suitable for sending
 *  records.
 *  @see SSF.Util.Streams
 */
public class ProbeSession extends SSF.OS.ProtocolSession {

  public int getHostCode() {
    if (str!=null)
      return str.getRecordSourceCode(((SSF.Net.Host)inGraph()).nhi);
    return -1;
  }

  public int getHostCode(int ifacenum) {
    if (str!=null)
      return str.getRecordSourceCode
             (((SSF.Net.Host)inGraph()).nhi+"("+ifacenum+")");
    return -1;
  }

  /**  Example DML configuration:
   *   <PRE>
   *   ProtocolSession [
   *     name probe use SSF.OS.ProbeSession
   *     file "flow_data/flowCampus2.dat"
   *     stream "ipnetflow"
   *   ]
   * </PRE>
  */
  public void config(Configuration cfg) throws configException {
    super.config(cfg);

    //-- added to handle wrapup of monitors.
    finalizers = new Vector();

    String file = (String)cfg.findSingle("file");

    String host = (String)cfg.findSingle("jahnu.host");
    String port = (String)cfg.findSingle("jahnu.port");

    if (file!=null && (host!=null || port!=null))
	    throw new configException("ProbeSession may not specify "+
				      "both file and host, port targets");

    if (file==null) {
      if (host==null) host="localhost";
      if (port==null) port="9000";
    }

    String streamname = (String)cfg.findSingle("stream");
      if (streamname==null)
        throw new configException("ProbeSession must contain 'stream'"+
		                  " attribute to identify stream");

	if (file!=null) 
	    config(file,streamname);
	else
          config(host,(new Integer(port)).intValue(),streamname);
    }

    public void config(String filename, String certif) 
	throws configException {

	// Get the alignment number X from "{Scheduler X}"
	String align = inGraph().alignment().toString();
	align = align.substring(align.indexOf(" ")).trim();
	align = align.substring(0,align.indexOf("}")).trim();

	certif = certif+"."+align;
	filename = filename+"."+align;

	System.err.println("Config: "+filename+" :: "+certif);
	try {
	    str = unique(filename,certif);
	} catch (streamException strex) {
	    throw new configException(strex.toString());
	}
	System.err.println("str == "+str);
    }

    public void config(String hostname, int portnum, String certif) 
	throws configException {

	// Get the alignment number X from "{Scheduler X}"
	String align = inGraph().alignment().toString();
	align = align.substring(align.indexOf(" ")).trim();
	align = align.substring(0,align.indexOf("}")).trim();

	certif = certif+"."+align;

	System.err.println("Config: "+hostname+":"+portnum+" :: "+certif);

	try {
	    str = unique(hostname,portnum,certif);
	} catch (streamException strex) {
	    throw new configException(strex.toString());
	}
	System.err.println("str == "+str);
    }

    private StreamInterface str;
    private static java.util.Hashtable table;
    private synchronized StreamInterface unique(String filename, 
						String ecertif) 
    throws streamException {
	if (table == null) table = new java.util.Hashtable();
	StreamInterface S = (StreamInterface)table.get(ecertif);
	if (S==null) {
	    S = new BasicRecorder(ecertif);
	    S.connectWrite(filename);
	    table.put(ecertif,S);
	}

	prepareWrapup(S);

	return S;
    }
    // changed to "non-static" to use "prepareWrapup"
    private synchronized StreamInterface unique(String hostname, 
						       int portnum, 
						       String ecertif) 
    throws streamException {
	if (table == null) table = new java.util.Hashtable();
	StreamInterface S = (StreamInterface)table.get(ecertif);
	if (S==null) {
	    S = new BasicRecorder(ecertif);
	    S.connectWrite("jahnu://"+hostname+":"+portnum);
	    table.put(ecertif,S);
	}

	prepareWrapup(S);

	return S;
    }

    private void prepareWrapup(StreamInterface sInterface)
    {
	final StreamInterface R = sInterface;

	((SSF.Net.Host)inGraph()).getNet().wrapup(new Runnable() {
		public void run() {
		    for (Enumeration fset = finalizers.elements();
			 fset.hasMoreElements();) 
			((Runnable)fset.nextElement()).run(); 
		    R.disconnect();
		}
	    });
	try {
	    ((BasicRecorder)(sInterface)).addSession();
	}
	catch (ClassCastException cce) {
	    System.err.println(cce.getMessage()+"Basic Player is expected.in ProbeSession.unique().");
	    cce.printStackTrace();
	    System.exit(-1);
	}
    }
    public boolean push(SSF.OS.ProtocolMessage msg, 
			SSF.OS.ProtocolSession from) {
	return false;
    }
    public StreamInterface getRecorder() { 
	try {
	    if (str==null) config(new dmlConfig());
	} catch (configException cfgex) {
	    System.err.println(cfgex);
	}
	return str; 
    }

    //-- added to deal with wrapup of monitors.
    private Vector finalizers;
    public StreamInterface GetRecorder(Runnable finalizer) {
	finalizers.addElement(finalizer);
	return getRecorder();
    }
}
